﻿using Confluent.Kafka;
using gt.EventBus;
using gt.EventBus.Abstractions;
using gt.EventBus.Events;
using gt.EventBus.Kafka;
using Newtonsoft.Json;
using System;
using System.Threading.Tasks;

namespace gt.eventbuskafka.sample
{
    class Program
    {
        const string _brokeList = "10.228.129.151:9092,10.228.129.152:9092,10.228.129.153:9092";
        static void Main(string[] args)
        {
            Console.WriteLine("start");

            var kafkaConnection = new DefaultKafkaPersistentConnection(new ProducerConfig
            {
                BootstrapServers = _brokeList,
                RequestTimeoutMs = 3000,
                MessageTimeoutMs = 10000,
                SecurityProtocol = SecurityProtocol.SaslPlaintext,
                SaslMechanism = SaslMechanism.Plain,
                SaslUsername = "kafka",
                SaslPassword = "kafka#secret"
            }, new ConsumerConfig
            {
                BootstrapServers = _brokeList,
                EnableAutoCommit = true,
                GroupId = "event-consumer",
                AutoOffsetReset = AutoOffsetReset.Latest,
                SecurityProtocol = SecurityProtocol.SaslPlaintext,
                SaslMechanism = SaslMechanism.Plain,
                SaslUsername = "kafka",
                SaslPassword = "kafka#secret"
            }, "EventBus_Test");

            //IEventBus eventBus = new InMemoryEventBus();
            IEventBus eventBus = new EventBusKafka(kafkaConnection);

            eventBus.Subscribe<CaseOneEvent, SimpleOneHandler>();
            eventBus.Subscribe<CaseOneEvent, SimpleOneOneHandler>();
            eventBus.Subscribe<CaseTwoEvent, SimpleTwoHandler>();


            eventBus.Publish(new CaseOneEvent { OneTag = "1111" });
            eventBus.Publish(new CaseOneEvent { OneTag = "1111-1111" });
            eventBus.Publish(new CaseTwoEvent { TwoTag = "2222" });
            eventBus.Publish(new CaseTwoEvent { TwoTag = "2222-2222" });


            Console.ReadLine();
        }
    }


    public class CaseOneEvent : IntegrationEvent
    {
        public string OneTag { get; set; }
    }
    public class CaseTwoEvent : IntegrationEvent
    {
        public string TwoTag { get; set; }
    }

    public class SimpleOneHandler : IIntegrationEventHandler<CaseOneEvent>
    {
        public Task Handle(CaseOneEvent @event)
        {
            Console.WriteLine($"this is simple one handler,eventMessage:{JsonConvert.SerializeObject(@event)}");
            return Task.CompletedTask;
        }
    }

    public class SimpleOneOneHandler : IIntegrationEventHandler<CaseOneEvent>
    {
        public Task Handle(CaseOneEvent @event)
        {
            Console.WriteLine($"this is simple one-one handler,eventMessage:{JsonConvert.SerializeObject(@event)}");
            return Task.CompletedTask;
        }
    }

    public class SimpleTwoHandler : IIntegrationEventHandler<CaseTwoEvent>
    {
        public Task Handle(CaseTwoEvent @event)
        {
            Console.WriteLine($"this is simple two handler,eventMessage:{JsonConvert.SerializeObject(@event)}");
            return Task.CompletedTask;
        }
    }
}
